Raspberry PiとAWSを利用して子どもたちのゲーム時間を可視化してみた

こんにちは、CX事業本部MAD事業部の森茂です。

突然ですが、我が家の子どもたちにとってゲームの時間はとても重要で、1日の制限時間は1秒たりともムダにできません。

おとな：もう3時間はゲームしてるんじゃない？
子ども：まだ2時間しかやっていない。
おとな：もう3時間過ぎているでしょう？もう終わりにしなさい。
子ども：ぴえん

というわけで、親子間のムダな争いを避けそれぞれの権利・主張・平和のために子どもたちが今日どれくらいゲームをやったのかをRaspberry PiとAWSを利用して可視化してみました。

はじめに

つくってみた構成はこちら。
若干省略していますが最終的にこのような形になりました。

システム構成

まず、Raspberry Pi ZeroWHを使ってゲームの開始、終了時間を測るタイマーボタンを設置。ボタンを押すごとにAPI Gateway経由でLambdaを起動、ゲームの開始時間、終了時間をDynamoDBに保存します。
開始時、終了時にはおとなのLINEに通知を行い、統計情報はS3にホスティングしたReactアプリケーションから閲覧できるようにします。オマケ機能としてEventBridgeとLambdaで毎夜合計時間を集計して、直近一週間の合計時間も閲覧できるようにしてみます。

利用ユーザー想定

今回の利用ユーザーはLucyとMikeの2名を想定。それぞれがゲームの開始時と終了時、自分に割り当てられたボタンを押します。

開発環境

• M1 MacBook Pro（Big Sur）
• Raspberry Pi ZeroWH
• Python 3.9.7
• node.js v16.10.0

バックエンド環境構築

バックエンドの環境構築にはCDK v2を使っていきます。今回はRaspberry Piを使うのでいつもならTypeScriptでまとめるところ、あまり書く機会のなかったPythonで挑戦してみました。

なお、CDKの利用にあたってAWS CLIの用意、また認証の設定などをあらかじめ完了しておく必要があります。

またLINE Notifyで通知を行う設計のためLINE Notifyトークンが必要になります。下記を参考にトークンの取得をしておきます。

CDK v2のインストール

今回はまだRC版ではありますがv2系を利用していきます。（RC23まできており個人用途であれば問題は少ないのではと考えています。） ※v1系でもライブラリのimportなどが若干異なりますがほぼ同じような構成で組み上げること可能です。

$ npm install -g cdk@next
$ cdk --version
2.0.0-rc.23 (build 1e54fb9)

CDKでスタックの雛形を作成

cdk init appスクリプトを利用してスタックの雛形を作ります。

$ cdk init app --language python
$ python3 -m venv .venv
$ source .venv/bin/activate # 以後は仮想環境内での作業
$ pip install -r requirements.txt

セットアップ後はこのような構成になりました。

├── README.md
├── app.py
├── cdk.json
├── game_counter
│   ├── __init__.py
│   ├── __pycache__
│   ├── game_counter.egg-info
│   └── game_counter_stack.py
├── requirements.txt
├── setup.py
└── source.bat

なお、CDKをはじめて利用する環境の場合、cdk bootstrapでAWS上に受け入れ環境を用意しておく必要があります。（v1系とは異なる受け入れ環境が作成されるためv2系をはじめ利用の場合はこの作業が必要です）

$ cdk bootstrap

追加ライブラリのインストール

今回アプリケーションでは外部ライブラリrequestsを利用するためインストールしておきます。 その他にもライブラリをimportしていますがLambdaのPython3.9環境にはあらかじめ用意されているライブラリのためローカルで動作確認を行う必要がなければ不要でした。

$ pip install requests

CDKでインフラ環境を構築

├── game_counter
│   └── game_counter_stack.py

今回はシンプルな構成のためファイルは分割せずgame_counter_stack.pyにすべてのリソースを記載していきます。

手始めにDynamoDBのスキーマについては下記のようにしてみました。

ログ記録用テーブル（GameCounter）

Raspberry Piからusername timestamp flagをAPI経由で渡す想定。

項目名	データ型	種別	備考
username	String	パーティションキー	名前（Lucy、Mike）
timestamp	Number	ソートキー	UNIXTIME（秒）
date	String		年月日（2021-10-10）
time	String		時間（12:00:00）
flag	String		開始、終了のフラグ（start | end）
uid	String		uuid（ID、機能拡張時に使うかも？）

uidについては今後の拡張で個別ログの編集や削除することを考えて用意しています。

集計用テーブル（GameCounterLog）

EventBridgeを利用して毎日ログを集計するためのテーブル。

項目名	データ型	種別	備考
username	String	パーティションキー	名前
date	String	ソートキー	年月日（2021-10-10）
totaltime	String		一日の合計時間（4:00:00）

DynamoDBの作成

さっそくテーブルをCDKで構築してみます。

from aws_cdk import (
    Stack,
    RemovalPolicy,
    aws_dynamodb as dynamodb, # DynamoDBのライブラリをimport
)
from constructs import Construct

class GameCounterStack(Stack):

    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # The code that defines your stack goes here
        # ここから下に追記していきます。

        # DynamoDB ログデータ格納用
        table = dynamodb.Table(
            self, "GameCounterTable",
            table_name="GameCounter",
            partition_key=dynamodb.Attribute(
                name="username", type=dynamodb.AttributeType.STRING),
            sort_key=dynamodb.Attribute(
                name="timestamp", type=dynamodb.AttributeType.NUMBER),
            billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
            removal_policy=RemovalPolicy.DESTROY
        )

        # DynamoDB 統計データ格納用
        log_table = dynamodb.Table(
            self, "GameCounterLogTable",
            table_name="GameCounterLog",
            partition_key=dynamodb.Attribute(
                name="username", type=dynamodb.AttributeType.STRING),
            sort_key=dynamodb.Attribute(
                name="date", type=dynamodb.AttributeType.STRING),
            billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
            removal_policy=RemovalPolicy.DESTROY
        )

この時点でいったんCDKの動作を確認してみます。
cdk synthで想定される構成の確認、エラーがないようであればcdk deployでデプロイします。

$ cdk synth
$ cdk deploy

念の為CLIからテーブルの一覧を確認。

$ aws dynamodb list-tables
{
    "TableNames": [
        "GameCounter",
        "GameCounterLog"
    ]
}

ここからは一気にいっちゃいます。 まずは必要なライブラリのimportまわりから。

from aws_cdk import (
    Stack,
    RemovalPolicy,
+    Duration,
+    CfnOutput,
    aws_dynamodb as dynamodb,
+    aws_lambda as lambda_,
+    aws_apigateway as apigateway,
+    aws_events as events,
+    aws_events_targets as targets
)
from constructs import Construct

さきほど作成したDynamoDBの記載下から一気にリソース（Lambda、API Gateway、S3、EventBridge）を追加します。

class GameCounterStack(Stack):

    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)
        
        ....

        # Lambda用共通パラメーター
        common_params = {
            "runtime": lambda_.Runtime.PYTHON_3_9,
            "timeout": Duration.seconds(10),
        }

        # Lambda データ一覧表示用
        list_data = lambda_.Function(
            self, "ListData",
            code=lambda_.Code.from_asset("lambda/list_data"),
            handler="index.handler",
            function_name="GameCounter-ListData",
            environment={
                "TABLE_NAME": table.table_name,
            },
            **common_params
        )

        # DynamoDBへ権限を付与
        table.grant_read_data(list_data)

        # SSMへのアクセス用のIAMロールを設定
        role_for_lambda = iam.Role(
            self, "Role",
            role_name="GameCounterSSMRole",
            assumed_by=iam.ServicePrincipal(
                "lambda.amazonaws.com"),
            managed_policies=[
                iam.ManagedPolicy.from_aws_managed_policy_name(
                    'service-role/AWSLambdaBasicExecutionRole'),
                iam.ManagedPolicy.from_aws_managed_policy_name(
                    'AmazonSSMReadOnlyAccess')
            ]
        )
	
        # Lambda データ投稿用
        push_data = lambda_.Function(
            self, "PushData",
            code=lambda_.Code.from_asset("lambda/push_data"),
            handler="index.handler",
            function_name="GameCounter-PushData",
	    # LINEトークンをSSMから取得するためにRoleを割り当て
	    role=role_for_lambda,
            environment={
                "TABLE_NAME": table.table_name,
            },
            **common_params
        )

        # DynamoDBへ権限を付与
        table.grant_write_data(push_data)

        # Lambda ログデータ統計用
        create_log = lambda_.Function(
            self, "CreateLog",
            code=lambda_.Code.from_asset("lambda/create_log"),
            handler="index.handler",
            function_name="GameCounter-CreateLog",
            environment={
                "TABLE_NAME": table.table_name,
                "LOG_TABLE_NAME": log_table.table_name,
            },
            **common_params
        )

        # DynamoDBへ権限を付与
        table.grant_read_data(create_log)
        log_table.grant_write_data(create_log)

        # Lambda データ一覧表示用
        list_log = lambda_.Function(
            self, "ListLog",
            code=lambda_.Code.from_asset("lambda/list_log"),
            handler="index.handler",
            function_name="GameCounter-ListLog",
            environment={
                "LOG_TABLE_NAME": log_table.table_name,
            },
            **common_params
        )

        # DynamoDBへ権限を付与
        log_table.grant_read_data(list_log)

        # ApiGatewayを作成
        api = apigateway.RestApi(
            self, "GameCounterApi",
            default_cors_preflight_options={
                "allow_origins": apigateway.Cors.ALL_ORIGINS,
                "allow_methods": apigateway.Cors.ALL_METHODS
            })

        # ApiGatewayのURI/counterというエンドポイントを作成
        counter_list = api.root.add_resource("counter")

        # /counter にGETでアクセスの際には一覧表示のLambdaを割り当てる
        counter_list.add_method(
            "GET", apigateway.LambdaIntegration(list_data),
            api_key_required=True
        )

        # /counter にPOSTでアクセスの際にはログ書き込みのLambdaを割り当てる
        counter_list.add_method(
            "POST", apigateway.LambdaIntegration(push_data),
            api_key_required=True
        )

        # 統計データ取得用のエンドポイント作成（counter/log）
        log_list = counter_list.add_resource("log")

        # /counter/log にGETでアクセスの際にはログ一覧表示のLambdaを割り当てる
        log_list.add_method(
            "GET", apigateway.LambdaIntegration(list_log),
            api_key_required=True
        )

        # API Keyを利用するために利用料プランとAPI KEYを作成
        plan = api.add_usage_plan(
            "UsagePlan",
            name="GameCounterPlan",
            throttle={
                "rate_limit": 10,
                "burst_limit": 2
            },
            api_stages=[{"api": api, "stage": api.deployment_stage}]
        )

        key = api.add_api_key(
            "GameCounterApiKey",
            api_key_name="GameCounterApiKey"
        )
        plan.add_api_key(key)

        # API Keyを確認するためにAPI KeyのIDをコンソールに出力
        CfnOutput(
            self, "GameCounterApiKeyExport",
                  value=key.key_id,
                  export_name="GameCounterApiKeyId"
        )

        # EventBridge 毎日24:30に前日の集計を行うLambdaを実行
        events.Rule(
            self, "GameCounterEventRule",
            rule_name="GameCounterEventRule",
            schedule=events.Schedule.cron(
                minute="30",
                hour="15",
                day="*"),
            targets=[targets.LambdaFunction(create_log, retry_attempts=3)]
        )

SSMにLINE Notify用トークンを格納

あらかじめ取得していたLINE Notify用のトークン情報をLambda上で利用するためSystems Managerのパラメーターストアに情報を入れておきます。

$ aws ssm put-parameter \
    --type 'SecureString' \
    --name '/GameCounter/LineToken' \
    --value 'LINE Notifyトークン情報'

Lambda関数の用意

プロジェクトルートにlambdaディレクトリを作成して用途別にlambda関数を用意します。create_logはEventBridgeから定期実行され、それ以外はAPI Gateway経由で実行してDynamoDBを操作するLambdaになります。
※バリデート処理など各種エラー処理は簡素化もしくは未実装のためご留意ください。またデータ量が少なく1回のクエリで全データ取得できる想定で、仮に取りこぼしても致命的な問題になるようなシステムではないため処理を一部省略しています。

./lambda
├── create_log # 統計ログ書き込み用
│   └── index.py
├── list_log   # 統計ログ一覧表示用
│   └── index.py
├── list_data  # ログ一覧表示用
│   └── index.py
└── push_data  # ログ書き込み用
    └── index.py

ログ書き込み用Lambda（lambda/push_data/index.py）

Raspberry PiからAPI Gateway経由で情報を受け取るLambda。requestsライブラリが別途必要なためあらかじめ同梱しておく必要があり。（今回はライブラリもそれを利用するLambdaも1つのみのためLayer化せずにそのまま入れています）

$ pip freeze | grep requests > lambda/push_data/requirements.txt
$ cd lambda/push_data
$ pip install -r requirements.txt -t .

import simplejson as json
import os
import uuid
import datetime
import boto3
from boto3.dynamodb.conditions import Key
import requests

dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ["TABLE_NAME"])

headers = {
    "Content-Type": "application/json",
    "Access-Control-Allow-Origin": "*",
}

# SSMパラメーターストアから値を取得
def get_parameters(param_key, ssm=None):
    if not ssm:
        ssm = boto3.client('ssm')

    response = ssm.get_parameter(
        Name=param_key,
        WithDecryption=True
    )
    return response['Parameter']['Value']


# UNIX時間をJSTのdatetimeに変換
def convert_timestamp(target):
    result = datetime.datetime.fromtimestamp(
        target, datetime.timezone(datetime.timedelta(hours=9)))
    return result


def handler(event, context):
    print("Received event: " + json.dumps(event))

    # クエリーパラメータの存在確認
    if event.get("body"):
        parsed = json.loads(event["body"])
        uid = str(uuid.uuid4())
        username = parsed["username"]
        timestamp = parsed["timestamp"]
        date = convert_timestamp(parsed["timestamp"]).date().isoformat()
        time = convert_timestamp(parsed["timestamp"]).time().isoformat()
        flag = parsed["flag"]
    else:
        username = None

    try:
        # usernameがある場合はDynamoDBにデータを登録
        if username:
            response = table.put_item(
                Item={
                    "uid": uid,
                    "username": username,
                    "timestamp": int(timestamp),
                    "date": date,
                    "time": time,
                    "flag": flag
                }
            )
            if response["ResponseMetadata"]["HTTPStatusCode"] != 200:
                raise Exception("Unable to connect to the database.")
            else:
                status_code = 200
                res = {"message": f"Created item with id: {uid}"}

            # LINEへ通知
            line_token = get_parameters("/GameCounter/LineToken")
            line_url = "https://notify-api.line.me/api/notify"
            line_headers = {"Authorization": "Bearer " + line_token}
            line_message = "flag未取得"

            if flag == "start":
                line_message = username + "がゲームを開始しました"
            if flag == "end":
                line_message = username + "がゲームを終了しました"

            payload = {"message": line_message}

            requests.post(
                line_url, headers=line_headers,
                params=payload
            )

        # usernameがない場合はエラーとして終了
        else:
            raise Exception("Parameter is missing.")

    except Exception as e:
        status_code = 500
        res = {"message": f"Internal Server Error. {str(e)}"}
    return {
        "statusCode": status_code,
        "headers": headers,
        "body": json.dumps(res)
    }

Raspberry Piからは以下のデータをJSON形式で受け取る想定です。

{
    "username": "Lucy", // String
    "timestamp: 1634113452, // Number unixtime（秒）
    "flag": "start" // start | end
}

ログ一覧表示用Lambda（lambda/list_data/index.py）

フロントエンド側からユーザー別、日付別のデータを取得するためのLambda。

import simplejson as json
import os
import boto3
from boto3.dynamodb.conditions import Key

dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ["TABLE_NAME"])

headers = {
    "Content-Type": "application/json",
    "Access-Control-Allow-Origin": "*",
}


def handler(event, context):
    print("Received event: " + json.dumps(event))

    # クエリーパラメータの存在確認
    if event.get("queryStringParameters"):
        username = event['queryStringParameters'].get("username")
        starttime = event['queryStringParameters'].get("starttime")
        endtime = event['queryStringParameters'].get("endtime")
    else:
        username = None

    try:
        # usernameの指定がある場合はusernameの指定した期間の一覧を出力
        if username:
            response = table.query(
                KeyConditionExpression=Key("username").eq(username) & Key(
                    "timestamp").between(int(starttime), int(endtime))
            )
        # usernameの指定がない場合は全件一覧を出力
        else:
            response = table.scan()

        if response["ResponseMetadata"]["HTTPStatusCode"] != 200:
            raise Exception("Unable to connect to the database.")
        else:
            status_code = 200
            res = response["Items"]
    except Exception as e:
        status_code = 500
        res = {"message": f"Internal Server Error. {str(e)}"}
    return {
        "statusCode": status_code,
        "headers": headers,
        "body": json.dumps(res)
    }

クエリパラメーターとして

• username…名前
• starttime…開始時間（unixtime）
• endtime…終了時間（unixtime）

を受け取ります。

統計ログ書き込み用Lambda（lambda/create_log/index.py）

EventBridgeから定期的に起動するLambda。GameCounterテーブルからデータを取得して合計時間をGameCounterLogテーブルに書き込みます。

import simplejson as json
import os
import boto3
from boto3.dynamodb.conditions import Key, Attr
import datetime

dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ["TABLE_NAME"])
log_table = dynamodb.Table(os.environ["LOG_TABLE_NAME"])


def query_today(username, target_date):
    response = table.query(
        KeyConditionExpression=Key("username").eq(username),
        FilterExpression=Attr("date").eq(target_date)
    )
    return response["Items"]


def create_total_time(log_data):
    total = 0

    for i, item in enumerate(log_data):
        temp = 0

        try:
            if log_data[i + 1].get("flag") == 'end':
                temp = int(log_data[i + 1]["timestamp"]) - \
                    int(item["timestamp"])
        except:
            print('done')

        total += temp

    return total


def handler(event, context):

    # 24時過ぎの計測を想定するため1日前の日付に変換
    today = str((datetime.datetime.now(
        datetime.timezone(datetime.timedelta(hours=9))) - datetime.timedelta(days=1)).date())

    try:

        # 対象の今日のログを集計、DBへ格納する
        targets = ["Lucy", "Mike"]

        for target in targets:
            res = query_today(target, today)
            total_time = create_total_time(res)
            data = str(datetime.timedelta(seconds=total_time or 0))

            if data:
                response = log_table.put_item(
                    Item={
                        "username": target,
                        "date": today,
                        "totaltime": data,
                    }
                )

    except Exception as e:
        print(f"Internal Server Error. {str(e)}")

    return "OK"

統計ログ一覧表示用Lambda（lambda/list_log/index.py）

フロントエンド側からユーザー別の統計情報を取得するためのLambda。

import simplejson as json
import os
import boto3
from boto3.dynamodb.conditions import Key

dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ["LOG_TABLE_NAME"])

headers = {
    "Content-Type": "application/json",
    "Access-Control-Allow-Origin": "*",
}


def handler(event, context):
    print("Received event: " + json.dumps(event))

    # クエリーパラメータの存在確認
    if event.get("queryStringParameters"):
        username = event['queryStringParameters'].get("username")
        target_date = event['queryStringParameters'].get("date")
    else:
        username = None

    try:
        # usernameの指定がある場合はusernameの指定した期間の一覧を出力
        if username:
            response = table.query(
                KeyConditionExpression=Key("username").eq(username) & Key(
                    "date").gte(target_date)
            )
        # usernameの指定がない場合は全件一覧を出力
        else:
            response = table.scan()

        if response["ResponseMetadata"]["HTTPStatusCode"] != 200:
            raise Exception("Unable to connect to the database.")
        else:
            status_code = 200
            res = response["Items"]
    except Exception as e:
        status_code = 500
        res = {"message": f"Internal Server Error. {str(e)}"}
    return {
        "statusCode": status_code,
        "headers": headers,
        "body": json.dumps(res)
    }

クエリパラメーターとして

• username…名前
• date…日付

を受け取ります。

CDKのデプロイ

ここまで一気に作ったところで環境をデプロイします。

$ cdk synth
$ cdk deploy

APIエンドポイント、API Keyの確認

デプロイ完了後、ターミナルコンソールにAPIのエンドポイントとAPIキーのIDが出力されることを確認してみます。

GameCounterStack.GameCounterApiEndpoint**** = https://****.execute-api.ap-northeast-1.amazonaws.cazonaws.com/prod/
GameCounterStack.GameCounterApiKeyExport = ********

エンドポイント末尾にcounterを追加したものが、一覧取得（GET）とログ登録（POST）に、末尾にcounter/logを追加したものが統計情報の一覧取得（GET）になります。

• https://xxxx.execute-api.ap-northeast-1.amazonaws.cazonaws.com/prod/counter
• https://xxxx.execute-api.ap-northeast-1.amazonaws.cazonaws.com/prod/counter/log

またAPIキーの確認のためターミナルからAWS CLIを使い下記コマンドで確認しておきます。APIキーはフロントエンド、Raspberry Piからそれぞれ利用します。

$ aws apigateway get-api-key \
    --api-key APIキーID \
    --include-value

{
    "id": "APIキーID",
    "value": "**APIキー**", #APIキーはフロントエンド、Raspberry Piから利用
    "name": "GameCounterApiKey",
    "enabled": true,
    "createdDate": "2021-10-10T15:18:17+09:00",
    "lastUpdatedDate": "2021-10-10T15:18:17+09:00",
    "stageKeys": [
        "xxxxxxxx/prod"
    ],
    "tags": {}
}

Raspberry Pi環境構築

やっとバックエンド周りが用意できました。次にボタンを押すとゲーム開始、さらに押すと終了の通知を送る装置をRaspberry Pi ZeroWHを使って構築します。今回はブレッドボードを利用してざっくりと装置をつくってみました。

装置のイメージ

タクトスイッチを押すとLEDが点灯し、もう一度押すとLEDが消灯する仕様。

用意したもの

• Raspberry Pi ZeroWH
• ブレッドボード
• LED x2個（Vf2.2Vくらい？）
• ジャンパーワイヤー（オス-オス、オス-メス）いくつか
• 抵抗（100Ω x2、 10kΩ x2）
• タクトスイッチ x2

10kΩの抵抗はプルアップ用に用意しています。Raspberry Piの本体側でも用意されているので、そちらを利用する場合は不要ですが今回は抵抗器を使って実装しています。

制御プログラム

装置側ができあがったところで、次に装置を制御するプログラムをRaspberry Piに用意していきます。
API Gatewayへボタンを押した際にリクエストを送るために外部ライブラリとしてrequestsをインストール。

$ pip install requests

import RPi.GPIO as GPIO
import time
import sys
import requests

# GPIOポートの設定
SwGpio = 24
Sw2Gpio = 6
LedGpio = 14
Led2Gpio = 5
lastStatus = 0
ledStatus = 0
last2Status = 0
led2Status = 0

# GPIOの設定
GPIO.setwarnings(False)
GPIO.setmode(GPIO.BCM)

# １つ目のボタン
GPIO.setup(SwGpio, GPIO.IN)
GPIO.setup(LedGpio, GPIO.OUT)

# 2つ目のボタン
GPIO.setup(Sw2Gpio, GPIO.IN)
GPIO.setup(Led2Gpio, GPIO.OUT)

# API情報 API GatewayのURLとAPI Keyを設定
API_URL = "https://xxxxxxx.execute-api.ap-northeast-1.amazonaws.com/prod/counter"
API_KEY = "****************"


def pushCount(name, flag):
    print("start")

    # UNIX時間（秒)で現在時間を取得
    timestamp = int(time.time())
    print(timestamp)

    url = API_URL
    headers = {"x-api-key": API_KEY}
    q = {'username': name, 'timestamp': timestamp, 'flag': flag}
    r = requests.post(url, headers=headers, json=q)
    print(r.json())


while True:
    try:
        btnStatus = not(GPIO.input(SwGpio))
        btn2Status = not(GPIO.input(Sw2Gpio))

        # 1つ目のボタンの動作を制御 今回は青いLEDをLucy用のボタンとします。
        if btnStatus == 0 and lastStatus == 1:
            ledStatus = not ledStatus
            if ledStatus == True:
                GPIO.output(LedGpio, True)
                pushCount('Lucy', 'start')
                print('button 1 ON')
            else:
                GPIO.output(LedGpio, False)
                pushCount('Lucy', 'end')
                print('button 1 OFF')
            # 連続でボタンを誤認識しないように少し待つ
            time.sleep(0.2)

        # ２つ目のボタンの動作を制御 今回は赤いLEDをMike用のボタンとします。
        if btn2Status == 0 and last2Status == 1:
            led2Status = not led2Status
            if led2Status == True:
                GPIO.output(Led2Gpio, True)
                pushCount('Mike', 'start')
                print('button 2 ON')
            else:
                GPIO.output(Led2Gpio, False)
                pushCount('Mike', 'end')
                print('button 2 OFF')
            # 連続でボタンを認識しないように少し待つ
            time.sleep(0.2)

        lastStatus = btnStatus
        last2Status = btn2Status

    except KeyboardInterrupt:
        GPIO.cleanup()
        sys.exit()

制御プログラムの用意ができたので、早速スクリプトを起動し、ボタンを押してデータを送信してみます。

$ python3 ./app.py

# ボタンを押すと開始
$ python3 game-counter/app.py

start
1634044409
{'message': 'Created item with id: a968f92a-7828-462a-b26a-bcbb3fab2810'}
button 2 ON

# もう一度ボタンを押すと終了

start
1634044423
{'message': 'Created item with id: 4c1c5fd5-e6b9-4bfc-b73b-11bcd2ae616d'}
button 2 OFF

# Ctrl + C で終了

これでボタンを押すたびにLEDが点灯してLINE Notify経由で通知が届くようになりました。

念の為、実際にログがDynamoDBに格納されているか確認してみます。

$ aws dynamodb scan --table-name GameCounter

{
    "Items": [
        {
            "date": {
                "S": "2021-10-12"
            },
            "flag": {
                "S": "start"
            },
            "timestamp": {
                "N": "1634044565"
            },
            "time": {
                "S": "22:16:05"
            },
            "username": {
                "S": "Lucy"
            },
            "uid": {
                "S": "4c1c5fd5-e6b9-4bfc-b73b-11bcd2ae616d"
            }
        },
        ...

ちゃんと入っていそう。

システムサービスへの登録

もろもろ動作を確認できたところでRaspberry Pi起動時にスクリプトを自動起動できるようsystemdにサービスとして登録します。

[Unit]
Description=GameCounterService
After=syslog.target

[Service]
Type=simple
# app.pyの格納先ディレクトリを指定 /home/pi/scripts/game-counter/app.pyに配置とした場合
WorkingDirectory=/home/pi/scripts/game-counter
ExecStart=/usr/bin/python3 /home/pi/scripts/game-counter/app.py
TimeoutStopSec=5
StandardOutput=null

[Install]
WantedBy = multi-user.target

systemdへ登録するために作成したcounter.serviceを/etc/systemd/system以下に格納してサービスの起動終了をテストしてみます。自動起動を設定することでRaspberry Piを起動するたびに自動的に動作するようになりました。

$ sudo cp counter.service /etc/systemd/system/
$ sudo systemctl daemon-reload

# 登録したサービスを起動
$ sudo systemctl start counter

# 念の為プロセスが動いているか確認
$ ps ax | grep app.py
263 ?        Rs   00:55 /usr/bin/python3 /home/pi/scripts/game-counter/app.py

# 登録したサービスを終了
$ sudo systemctl stop counter

# 起動、終了に問題なければ自動起動を設定
$ sudo systemctl enable counter

フロントエンド構築

長丁場になってきましたが、最後にフロントエンドの構築。 今回はAPIとのやり取りにRedux ToolkitのRTK Queryが便利（と個人的に推しているだけ）なのでcreate-react-appのreduxのテンプレートを使って構築します。 なお、UIにはChakra UIを、日時の計算用にmoment、ページ遷移のためにReactRouterを利用しました。

$ yarn create react-app frontend --template redux-typescript
$ cd frontend
$ yarn add @chakra-ui/react @emotion/react@^11 @emotion/styled@^11 framer-motion@^4
$ yarn add moment
$ yarn add react-router-dom
$ yarn add -D @types/react-router-dom

また、API情報については環境変数ファイルを用意してAPI GatewayのURLとAPI Keyを記載しておきます。

REACT_APP_API_URL=https://********.execute-api.ap-northeast-1.amazonaws.com/prod/counter
REACT_APP_API_KEY=********

ページ構成

今回はホーム画面と詳細画面を実装していきます。

• ホーム画面…LucyとMikeの直近7日間の合計時間一覧を表示
• ユーザー別詳細画面…ユーザー別の当日のゲーム時間のログ一覧を表示

フロントエンド実装

Reduxのテンプレートを使用しているので追加、変更したファイルは以下のようになりました。（詳細のソースはGitHubより参照ください）

frontend/
+── .env.local
├── README.md
├── package.json
├── public
├── src
│   ├── App.css
│   ├── App.test.tsx
+   ├── App.tsx
│   ├── app
│   │   ├── hooks.ts
+   │   └── store.ts
+   ├── components
+   │   └── Header.tsx
│   ├── features
│   │   └── counter
│   ├── index.css
+   ├── index.tsx
│   ├── logo.svg
+   ├── pages
+   │   ├── Home.tsx
+   │   ├── Page404.tsx
+   │   └── UserPage.tsx
│   ├── react-app-env.d.ts
│   ├── serviceWorker.ts
+   ├── services
+   │   └── game-counter.ts
│   ├── setupTests.ts
+   └── types
+       └── index.ts
├── tsconfig.json
└── yarn.lock

RTK Queryの紹介

少し脇道に逸れますが、フロントエンド側でAPIからデータを取得するためにRedux ToolkitのRTK Queryを利用しています。RTK QueryはRedux Toolkit v1.5から導入された新機能でswrやreact-queryと同じようなキャッシュ戦略を駆使したRedux Toolkitチーム謹製のQueryツールとなっています。

下記は統計ログ一覧の取得、ユーザーごとのログの取得のRTK Query部分になりますが、この記述のみで取得やエラーハンドリングなど複数の機能をもったカスタムフックを生成してくれます。

import { createApi, fetchBaseQuery } from '@reduxjs/toolkit/query/react';
import { GameCounterItem, GameCounterLogItem, GetQueryParams, LogQueryParams } from '../types';

// .env.localファイルからAPI URLとAPI KEYを取得
const API_URL = process.env.REACT_APP_API_URL!;
const API_KEY = process.env.REACT_APP_API_KEY!;

// RTK Queryを利用したAPIを作成
export const gameCounterApi = createApi({
  reducerPath: 'gameCounterApi',
  baseQuery: fetchBaseQuery({
    baseUrl: API_URL,
    // headerにAPI KEYを付与
    prepareHeaders: (headers) => {
      headers.set('x-api-key', API_KEY);
      return headers;
    },
  }),
  endpoints: (builder) => ({
    // ユーザー詳細画面でユーザーごとの情報を取得するためのクエリー
    // ユーザ名と期間（UNIXTIME）を送る
    getCounterByName: builder.query<GameCounterItem[], GetQueryParams>({
      query: (params) =>
        `?username=${params.username}&starttime=${params.startTime}&endtime=${params.endTime}`,
    }),
    // ホーム画面でユーザーごとの統計情報を取得するためのクエリー
    // ユーザ名と取得日（YYYY-MM-DD）を送る
    getLogDataByName: builder.query<GameCounterLogItem[], LogQueryParams>({
      query: (params) => `log?username=${params.username}&date=${params.date}`,
    }),
  }),
});

// RTK Queryが自動的に用意してくれる！カスタムフックをエクスポート
export const { useGetCounterByNameQuery, useGetLogDataByNameQuery } = gameCounterApi;

コンポーネント側からは、エクスポートされたuseGetCounterByNameQueryというカスタムフックを受け取ることで、下記複数のプロパティを返してくれます。

property
data	結果を返す
error	エラー結果を返す
isUninitialized	実行前にtrueを返す
isLoading	最初の読み込み中にtrueを返す
isFetching	最初だけでなく、以後再読み込み中のときもtrueを返す
isSuccess	リクエストが成功時にtrueを返す
isError	エラー時にtrueを返す
refetch	キャッシュを使わずに再読込する関数

主に利用するのは、data isLoading isFetchingが多そうです。

...

const { data, isLoading, isFetching } = useGetCounterByNameQuery(params, {
    // 30000秒ごとに自動更新とかもできます
    pollingInterval: 30000,
});

...

// 最初の読み込み時はLoadingを表示
if (isLoading) return <Box p={4}>Loading...</Box>;

// データが取得できなかった場合にエラーを表示
if (!data) return <Box p={4}>Missing data!</Box>;

// 万事うまくいった場合にコンテンツを表示
return (
      // 再読み込み中のときにSpinnerなどローディング状態を表示
      {isFetching && (
        <Box alignItems="center" justifyContent="center" textAlign="center">
          <Spinner thickness="4px" speed="0.65s" emptyColor="gray.200" color="blue.500" size="xl" />
        </Box>
      )}
      ...コンテンツを表示
)
...

RTK Queryを利用することで機能の役割をシンプルにわけることができるのと、Redux DevToolsでデバッグできる点など開発のメリットも多数あるかと思います。swrやreact-queryとはまた別の選択肢のひとつとしてぜひ試してみてください。

少し古いのですが以前RTK Queryのさわりを紹介した記事になりますのでよろしければこちらも参照ください。

フロントエンドのデプロイ

今回はフロントエンドのデプロイ先としてS3を利用することにしました。CDKを利用してS3バケットの作成とS3バケットへのデプロイを追記します。

from aws_cdk import (
    Stack,
    RemovalPolicy,
    Duration,
    CfnOutput,
    aws_dynamodb as dynamodb,
    aws_lambda as lambda_,
    aws_iam as iam,
    aws_apigateway as apigateway,
+    aws_s3 as s3,
+    aws_s3_deployment as s3deploy,
    aws_events as events,
    aws_events_targets as targets
)
from constructs import Construct

class GameCounterStack(Stack):

    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs

        ....

        # S3 Bucket フロントエンド公開用
        bucket = s3.Bucket(
            self, "GameCounterBucket",
            website_index_document="index.html",
            website_error_document="index.html",
            public_read_access=True,
            removal_policy=RemovalPolicy.DESTROY
        )

        # S3 BucketにReactからビルドしたファイルをアップロード
        s3deploy.BucketDeployment(
            self, "GameCounterBucketDeployment",
            destination_bucket=bucket,
            sources=[s3deploy.Source.asset("frontend/build")],
            retain_on_delete=False
        )

        # S3 Bucketの公開URLを出力
        CfnOutput(
            self, "GameCounterS3PublickUrl",
                  value=bucket.bucket_website_url,
                  export_name="GameCounterS3PublicUrl"
        )

S3へはReactのbuildディレクトリをアップロードするため、CDKをデプロイする前にReactアプリケーションをビルドしておきます。

$ cd frontend/
$ yarn build
$ cd ../
$ cdk synth
$ cdk deploy

...
GameCounterStack.GameCounterS3PublickUrl = http://gamecounterstack-gamecounterbucket*********.s3-website-ap-northeast-1.amazonaws.com

デプロイ終了時に出力されたS3バケットの公開URLへアクセスすることで無事にアプリケーションのホーム画面が表示されました。

おかたづけ

リソースが不要になった場合に備えてリソースの削除も試しておきます。※S3バケット、DynamoDBのRemovalPolicyをDESTROYとしているためすべてのデータが消えるので注意。

$ cdk destroy

SSMへ登録したLINEトークンは下記で削除。

$ aws ssm delete-parameter --name "/GameCounter/LineToken"

さいごに

かなりの長文になってしまいましたが、ここまで読んでいただきありがとうございました。

今回はRaspberry PiとAWSでよく使われる技術をざっくりと組み合わせてこどもたちを巻き込めるアプリケーションをつくってみました。週末に家庭内でのネタから突然つくりあげたところ、子どもたちにも受けがよくそのまま採用され絶賛運用稼働中です。

また、最近のデジタルネイティブな子どもたちはPythonということばを知っていたようでRaspberry Piをつかった電子工作やプログラミングにも興味をもってくれたことが新たな発見でした。

Raspberry PiやAWSを使ってみたいけど何をしようか、という方にも何かをつくるきっかけになれば幸いです。

完成版のソースについてはGitHubに公開しています。ハードコーディングな部分を最適化したり、グラフ表示を入れたりと拡張すべきところはまだまだたくさんありそうです。（手抜きな部分は徐々にアップデートしようと思っています。。。）

ん、話と違って、なんか子どもたちのゲーム時間長すぎ？！